package com.heima.article.stream;

import com.heima.common.constants.CommentComstants;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.time.Duration;

@Configuration
@Slf4j
public class ArticleCommentStreamHandle {

    @Bean("kStreamOfComment")
    public KStream<String, String> kStreamOfComment(StreamsBuilder streamsBuilder) {
        //创建出kStream
        KStream<String, String> stream = streamsBuilder.stream(CommentComstants.ARTICLE_COMMENT_TOPIC);
        //流式处理
        stream.map((key, value) -> {
            long articleId = Long.parseLong(value);
            return new KeyValue<>(articleId, 1);
        }).groupBy((key, value) -> value)
                .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
                //自行聚合数据
                .aggregate(new Initializer<Integer>() {
                    @Override
                    public Integer apply() {
                        return 0;
                    }
                }, new Aggregator<Integer, Integer, Integer>() {
                    @Override
                    public Integer apply(Integer key, Integer value, Integer app) {
                        return value + app;
                    }
                }, Materialized.as("article-comment-001"))
                .toStream()
                .map((key,value) ->{
                    return new KeyValue<>(key.key().toString(),key.key()+":"+value);
                }).to(CommentComstants.ARTICLE_COMMENT_HANDLE_TOPIC);

        return stream;
    }
}
